package sql;

import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

/**

 

 */
public class D7_KafakSource_rownumber_redis_state {


    public static void main(String[] args) {


        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        // flinkConf.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, new MemorySize(128 * 1024 * 1024);
        flinkConf.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, new MemorySize(512 * 1024 * 1024));
        flinkConf.set(TaskManagerOptions.TASK_HEAP_MEMORY, new MemorySize(128 * 1024 * 1024));
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);


        // env.enableCheckpointing(2000);

        // rocksDB
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("file:///flink_ckp");



        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // String genSql = "CREATE TABLE ods_tb ( " +
        //         " stime STRING," +
        //         " name STRING," +
        //         " id BIGINT," +
        //         " rowtime AS to_timestamp(stime)," +
        //         " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
        //         ") WITH ( " +
        //     "  'connector' = 'kafka'," +
        //     "  'topic' = 'test'," +
        //     "  'properties.bootstrap.servers' = 'wsl:9092'," +
        //     "  'properties.group.id' = 'testGroup'," +
        //     "  'scan.startup.mode' = 'latest-offset'," +
        //     "  'format' = 'json'" +
        //         ")";


        String genSql = "CREATE TABLE ods_tb ( " +
            " stime TIMESTAMP(3)," +
            " name STRING," +
            " id BIGINT," +
            " rowtime AS localtimestamp," +
            " WATERMARK FOR rowtime AS rowtime - interval '0' second" +
            ") WITH ( " +
            " 'connector'='datagen', " +
            " 'rows-per-second'='100'," +
            " 'fields.name.length'='100'," +
            " 'fields.id.kind'='random'" +
            ")";

        String print = "CREATE TABLE redis (" +
            "    snkey STRING, " +
            "    key STRING, " +
            "    val STRING, " +
            " primary key (snkey) not ENFORCED" +
            ") WITH (" +
            "    'connector' = 'redis',\n" +
            "    'host' = 'bdyun-suzhou-tw04f0163',\n" +
            "    'port' = '6379',\n" +
            "    'redis-mode' = 'single',\n" +
            "    'timeout' = '2000',\n" +
            "    'password' = 'yy2024',\n" +
            "    'command' = 'HMSET',\n" +
            "    'max.retries' = '5',\n" +
            "    'database' = '5',\n" +
            "    'ttl' = '10800'" +
            ")";

        String sql = " CREATE VIEW vv AS " +
                " SELECT " +
            " stime," +
            " name," +
            " id," +
            " DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd HH:mm:ss') nowtime" +
            " FROM " +
            " (" +
            " SELECT *,ROW_NUMBER() OVER(PARTITION BY name ORDER BY rowtime ASC) as rownum FROM ods_tb" +
            " )  where rownum = 1";




        String sinkSql = "INSERT INTO redis SELECT " +
            " name as snkey," +
            " nowtime as key," +
            "'aa' as val" +
            " FROM vv";


        tableEnv.executeSql(genSql);
        tableEnv.executeSql(print);

        System.out.println(sql);
        tableEnv.executeSql(sql);
        tableEnv.executeSql(sinkSql);


        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
